package com.atguigu.bigdata.spark.core.rdd.operator.transform;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

public class Spark16_RDD_Operator_Transform_JAVA {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        JavaSparkContext sc = new JavaSparkContext(conf);
        // TODO 算子 - (Key - Value类型)
        // reduceByKey : 相同的key的数据进行value数据的聚合操作
        List<Tuple2<String, Integer>> list = Arrays.asList(new Tuple2<String,Integer>("1", 1),
                new Tuple2<String,Integer>("1",2), new Tuple2<String, Integer>("2", 2),
                new Tuple2<String, Integer>("1", 3), new Tuple2<String, Integer>("2", 4));
        JavaRDD<Tuple2<String, Integer>> rdd = sc.parallelize(list, 2);
        JavaPairRDD<String, Integer> pairRDD = rdd.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return new Tuple2<String, Integer>(stringIntegerTuple2._1,stringIntegerTuple2._2);
            }
        });
        // groupByKey : 将数据源中的数据，相同key的数据分在一个组中，形成一个对偶元组
        //              元组中的第一个元素就是key，
        //              元组中的第二个元素就是相同key的value的集合
        //数据打乱重组，存在shuffle操作 落盘进行处理
        JavaPairRDD<String, Iterable<Integer>> res  = pairRDD.groupByKey();

        System.out.println(res.collect().toString());

        sc.stop();

    }
}
